package com.muyan.sys.webscoket.server;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson2.JSON;
import com.muyan.sys.common.core.domain.model.LoginUser;
import com.muyan.sys.common.core.redis.RedisCache;
import com.muyan.sys.common.utils.SecurityUtils;
import com.muyan.sys.framework.web.service.TokenService;
import com.muyan.sys.webscoket.constant.ChatCacheConstants;
import com.muyan.sys.webscoket.protocol.MessageProto;
import com.muyan.sys.webscoket.service.ChatGroupService;
import com.muyan.sys.webscoket.service.ServerInstanceManager;
import com.muyan.sys.webscoket.util.ChannelManager;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class WebSocketMessageHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    private final RabbitTemplate rabbitTemplate;
    private final ChannelManager channelManager;
    private final ServerInstanceManager serverInstanceManager;
    private final RedisCache redisCache;
    private final ChatGroupService chatGroupService;


    private final TokenService tokenService;

    public WebSocketMessageHandler(RabbitTemplate rabbitTemplate,
                                   ChannelManager channelManager,
                                   ServerInstanceManager serverInstanceManager,
                                   RedisCache redisCache,
                                   ChatGroupService chatGroupService, TokenService tokenService) {
        this.rabbitTemplate = rabbitTemplate;
        this.channelManager = channelManager;
        this.redisCache = redisCache;
        this.chatGroupService = chatGroupService;
        this.serverInstanceManager = serverInstanceManager;
        this.tokenService = tokenService;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) {
        String text = frame.text();
        try {
            log.info("消息处理={}", text);
            MessageProto message = JSON.parseObject(text, MessageProto.class);

            handleMessage(ctx.channel(), message);

            LoginUser user = (LoginUser) ctx.channel().attr(io.netty.util.AttributeKey.valueOf("LoginUser")).get();
            tokenService.verifyToken(user);
        } catch (Exception e) {
            log.error("消息处理异常", e);
        }
    }

    private void handleMessage(Channel channel, MessageProto message) {
        switch (message.getType()) {
            case CONNECT:
                handleConnect(channel);
                break;
            case UNREADGROUP:
                LoginUser user = (LoginUser) channel.attr(io.netty.util.AttributeKey.valueOf("LoginUser")).get();
                String userId = user != null ? user.getUserId().toString() : "";
                String userStatusKey = ChatCacheConstants.USER_STATUS_KEY + userId;
                String currentGroupId = redisCache.getCacheObject(userStatusKey);
                sendUnreadGroupMessages(userId, currentGroupId, channel);
                break;
            case CHAT:
                handleChat(message);
                break;
            case GROUP_LIST:
                handleGroupList(message);
                break;
            case GROUP_CHAT:
                handleGroupChat(message);
                break;
            case HEARTBEAT:
                handleHeartbeat(channel);
                break;
            case DISCONNECT:
                handleDisconnect(channel, message);
                break;
            case USER_STATUS:
                handleUserStatus(message);
                MessageProto updateMessage = new MessageProto();
                updateMessage.setType(MessageProto.MessageType.USER_STATUS);

                channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(updateMessage)));
                break;
        }
    }

    private void handleGroupList(MessageProto message) {
        String userId = message.getFromUserId();
        String groupId = message.getGroupId();

        // 获取用户当前所在的群组ID
        String userStatusKey = ChatCacheConstants.USER_STATUS_KEY + userId;
        String currentGroupId = redisCache.getCacheObject(userStatusKey);

        // 如果用户不在任何群聊中，或者不在当前群聊中
        if (currentGroupId == null || !currentGroupId.equals(groupId)) {
            // 获取未读消息数
            String unreadKey = ChatCacheConstants.GROUP_UNREAD_KEY + userId + ":" + groupId;
            Long unreadCount = redisCache.getCacheObject(unreadKey);

            if (unreadCount != null && unreadCount > 0) {
                // 发送群组列表更新消息
                MessageProto updateMessage = new MessageProto();
                updateMessage.setType(MessageProto.MessageType.GROUP_LIST);
                updateMessage.setGroupId(groupId);
                updateMessage.setContent("未读消息: " + unreadCount);

                Channel userChannel = channelManager.getChannel(userId);
                if (userChannel != null && userChannel.isActive()) {
                    userChannel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(updateMessage)));
                }
            }
        }
    }

    private void handleConnect(Channel channel) {
        log.info("用户连接id----------------: {}", channel.id());
        LoginUser user = (LoginUser) channel.attr(io.netty.util.AttributeKey.valueOf("LoginUser")).get();
        String userId = user != null ? user.getUserId().toString() : "";
        channelManager.addChannel(userId, channel);
        sendUnreadMessages(userId, channel);
        log.info("用户连接: {}", userId);
        log.info("获取用户连接: {}", channelManager.getChannel(userId));
    }

    private void handleUserStatus(MessageProto message) {
        String userId = message.getFromUserId();
        String currentGroupId = message.getGroupId(); // 当前所在的群组ID

        // 更新用户当前所在的群组ID
        String userStatusKey = ChatCacheConstants.USER_STATUS_KEY + userId;
        if (currentGroupId != null && !currentGroupId.isEmpty()) {
            redisCache.setCacheObject(userStatusKey, currentGroupId);
        } else {
            redisCache.deleteObject(userStatusKey);
        }

        log.info("更新用户状态: userId={}, currentGroupId={}", userId, currentGroupId);
    }

    private void handleChat(MessageProto message) {
        String toUserId = message.getToUserId();

        // 检查用户是否在当前服务器
        Channel toChannel = channelManager.getChannel(toUserId);
        if (toChannel != null && toChannel.isActive()) {
            // 用户在当前服务器，直接发送消息
            toChannel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(message)));
        } else {
            // 检查用户是否在其他服务器
            String userServerId = channelManager.getUserServerId(toUserId);
            if (userServerId != null) {
                // 用户在其他服务器，通过RabbitMQ发送消息
                rabbitTemplate.convertAndSend(
                        ChatCacheConstants.MQ_PRIVATE_EXCHANGE,
                        JSON.toJSONString(message)
                );
            } else {
                // 用户不在线，可以存储为离线消息
                List<MessageProto> list = new ArrayList<>();
                list.add(message);
                redisCache.setCacheList(ChatCacheConstants.QUE_PRIVATE + toUserId, list);
                //离线消息最多存一年
                redisCache.expire(ChatCacheConstants.QUE_PRIVATE + toUserId, 365, TimeUnit.DAYS);
            }
        }
    }

    private void handleGroupChat(MessageProto message) {
        String groupId = message.getGroupId();
        String fromUserId = message.getFromUserId();

        // 获取群组成员
        Set<String> groupMembers = chatGroupService.getGroupMembers(groupId);
        log.info("groupId={}", groupId);
        log.info("groupMembers={}", JSON.toJSONString(groupMembers));

        if (groupMembers == null) {
            return;
        }

        // 转发消息给群组在线成员
        for (String userId : groupMembers) {
            log.info("群组用户id={}",userId);
            if (userId.equals(fromUserId)) {
                continue; // 跳过发送者自己
            }

            // 检查用户是否在线
            if (channelManager.isOnline(userId)) {
                // 获取用户当前所在的群组ID
                String userStatusKey = ChatCacheConstants.USER_STATUS_KEY + userId;
                String currentGroupId = redisCache.getCacheObject(userStatusKey);
                log.info("获取用户当前所在的群组ID={}={}",userId, currentGroupId);
                // 如果用户当前就在这个群聊中，直接推送消息
                if (groupId.equals(currentGroupId)) {
                    Channel userChannel = channelManager.getChannel(userId);

                    if (userChannel != null && userChannel.isActive()) {//在当前服务器
                        log.info("发送消息成功={}={}={}",userId, currentGroupId,message.getContent());
                        userChannel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(message)));
                    } else {
                        log.info("不在当前服务器中={}={}", userId,currentGroupId);
                        //用户在其他服务器，通过RabbitMQ发送消息
                        rabbitTemplate.convertAndSend(
                                ChatCacheConstants.MQ_GROUP_CHAT_EXCHANGE,
                                JSON.toJSONString(message)
                        );
                    }
                } else {
                    log.info("群组用户id不在群组中={}={}={},",userId,groupId,currentGroupId);
                    // 如果用户不在这个群聊中，增加未读消息计数
                    String unreadKey = ChatCacheConstants.GROUP_UNREAD_KEY + userId + ":" + groupId;
                    redisCache.increment(unreadKey);
                    redisCache.expire(unreadKey, 30, TimeUnit.DAYS); // 设置30天过期时间

                    // 发送群组列表更新消息
                    MessageProto updateMessage = new MessageProto();
                    updateMessage.setType(MessageProto.MessageType.GROUP_LIST);
                    updateMessage.setGroupId(groupId);
                    updateMessage.setContent("新消息");

                    Channel userChannel = channelManager.getChannel(userId);
                    if (userChannel != null && userChannel.isActive()) {
                        userChannel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(updateMessage)));
                    }
                    //不在群聊中保存消息
                    String offlineKey = ChatCacheConstants.OFFLINE_GORUP_MESSAGES_KEY + userId + ":" + groupId;

                    List<MessageProto> offlineMessages = redisCache.getCacheList(offlineKey);
                    if (offlineMessages == null) {
                        offlineMessages = new ArrayList<>();
                    }
                    offlineMessages.add(message);
                    redisCache.setCacheList(offlineKey, offlineMessages);
                    redisCache.expire(offlineKey, 30, TimeUnit.DAYS);
                }
            } else {
                log.info("用户不在线={}", userId);
                // 用户不在线，存储为离线消息

                String offlineKey = ChatCacheConstants.OFFLINE_GORUP_MESSAGES_KEY + userId + ":" + groupId;

                List<MessageProto> offlineMessages = redisCache.getCacheList(offlineKey);
                if (offlineMessages == null) {
                    offlineMessages = new ArrayList<>();
                }
                offlineMessages.add(message);
                redisCache.setCacheList(offlineKey, offlineMessages);
                redisCache.expire(offlineKey, 30, TimeUnit.DAYS);
            }
        }
    }

    private void sendUnreadMessages(String userId, Channel channel) {
        // 获取离线消息
        String offlineKey = ChatCacheConstants.OFFLINE_MESSAGES_KEY + userId;
        List<MessageProto> offlineMessages = redisCache.getCacheList(offlineKey);
        if (offlineMessages != null && !offlineMessages.isEmpty()) {
            for (MessageProto message : offlineMessages) {
                channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(message)));
            }
            redisCache.deleteObject(offlineKey);
        }

        // 获取未读消息计数
        String unreadPattern = ChatCacheConstants.GROUP_UNREAD_KEY + userId + ":*";
        Set<String> unreadKeys = redisCache.keys(unreadPattern);
        if (unreadKeys != null && !unreadKeys.isEmpty()) {
            for (String key : unreadKeys) {
                String groupId = key.substring(key.lastIndexOf(":") + 1);
                Integer count = redisCache.getCacheObject(key);
                if (count != null && count > 0) {
                    MessageProto updateMessage = new MessageProto();
                    updateMessage.setType(MessageProto.MessageType.GROUP_LIST);
                    updateMessage.setGroupId(groupId);
                    updateMessage.setContent("未读消息: " + count);
                    channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(updateMessage)));
                }
            }
        }
    }

    private void sendUnreadGroupMessages(String userId, String groupId, Channel channel) {
        // 获取离线消息

        String offlineKey = ChatCacheConstants.OFFLINE_GORUP_MESSAGES_KEY + userId + ":" + groupId;
        // 获取未读消息计数
        List<MessageProto> offlineMessages = redisCache.getCacheList(offlineKey);
        if (CollectionUtil.isNotEmpty(offlineMessages)) {
            for (MessageProto message : offlineMessages) {
                channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(message)));
            }
            redisCache.deleteObject(offlineKey);
        }


    }

    private void handleHeartbeat(Channel channel) {
        // 更新服务器心跳
        LoginUser user = (LoginUser) channel.attr(io.netty.util.AttributeKey.valueOf("LoginUser")).get();
        String userId = user != null ? user.getUserId().toString() : "";
        serverInstanceManager.updateHeartbeat(channel, userId);
        MessageProto updateMessage = new MessageProto();
        updateMessage.setType(MessageProto.MessageType.HEARTBEAT);

        channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(updateMessage)));
    }

    private void handleDisconnect(Channel channel, MessageProto message) {
        String userId = message.getFromUserId();
        channelManager.removeChannel(userId);
        channel.close();
        log.info("用户断开连接: {}", userId);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.READER_IDLE) {
                // 60秒没有收到消息，关闭连接
                ctx.close();
            }
        }
        super.userEventTriggered(ctx, evt);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("WebSocket异常", cause);
        ctx.close();
    }
} 